package server

import (
	"core/datetime"
	"core/db"
	"core/id"
	"core/kafka"
	"core/redis"
	"core/server/baseserver"
	"iotall/iotcenter/centerconfig"
	"iotall/iotcore/entity"
	"strconv"
	"time"

	"github.com/dubbogo/gost/log/logger"
	kafkago "github.com/segmentio/kafka-go"
	"gorm.io/gorm/clause"
)

type CommDeviceOnlineDto struct {
	Id           int64 `json:"id,string"                         gorm:"column:id;primaryKey;autoIncrement:false;comment:主键;"`
	CommDeviceId int64 `json:"commDeviceId,string"                         gorm:"column:comm_device_id;comment:设备编号;"`

	OnlineDt       string    `json:"onlineDt"                         gorm:"column:online_dt;size:20;comment:上线时间;"`
	OffDt          string    `json:"offDt"                         gorm:"column:off_dt;size:20;comment:下线时间;"`
	Seconds        int       `json:"seconds"                         gorm:"column:seconds;comment:在线秒数;"`
	LogicTimeOut   int32     `json:"logicTimeOut"                         gorm:"column:logic_time_out;  DEFAULT:0;comment:逻辑离线时间;"`
	LstRevDt       time.Time `json:"lstRevDt"                         gorm:"column:lst_rev_dt;comment:最近上传数据时间;"`
	CpLogicTimeOut int32     `json:"cplogicTimeOut"                         gorm:"column:cp_logic_time_out;  DEFAULT:0;comment:逻辑离线时间;"`
}

func (p *CommDeviceOnlineDto) TableName() string {
	return "CommDeviceOnline"
}

type OnlineServer struct {
	baseserver.BaseServer
	reader *kafkago.Reader
}

func (p *OnlineServer) Start() {
	p.reader = kafka.GetKafkaReader(entity.EVENTTOPIC, "online")
	p.Name = "设备在线状态"
	p.Interval = 5000
	p.BaseServer.DoInterval = p.DoInterval
	p.BaseServer.Start()
}
func (p *OnlineServer) Close() {

}
func (p *OnlineServer) DoInterval() {
	var b, err = Zklocker.Lock(entity.ONLINE_LOCK_PATH, strconv.FormatInt(centerconfig.Config.Id, 10))
	if err != nil {
		return
	}
	if !b {
		return
	}
	dbnow, err := db.DbMasterNow()
	if err != nil {
		return
	}

	var cachetime = time.Time{}
	err = redis.RedisGetStruct("cache:OnlineServer:lstDt", &cachetime)
	if err != nil {
		cachetime = time.Now().Add(time.Hour * 1)
	}

	err = p.Online(dbnow, cachetime)
	if err != nil {
		return
	}
	err = redis.RedisSetStruct("cache:OnlineServer:lstDt", &dbnow, 0)
	if err != nil {
		return
	}

	p.Offline(dbnow)
}
func (p *OnlineServer) Online(dbnow, cachetime time.Time) error {

	var list []entity.CommDevice
	tx := db.DbMaster.Model(&entity.CommDevice{}).
		Joins("left join CommDeviceOnline b on CommDevice.id =b.comm_device_id and b.online_dt=''").
		Where("CommDevice.logic_state!=?", "在线").
		Where("CommDevice.lst_rev_dt>=?", cachetime).
		Where("CommDevice.lst_rev_dt<?", dbnow).
		Where("b.id is   null ").
		Select("CommDevice.id,CommDevice.lst_rev_dt").
		Find(&list)
	if tx.Error != nil {
		logger.Error(tx.Error)
		return tx.Error
	}
	var ids []int64
	var dataonline []entity.CommDeviceOnline
	if len(list) == 0 {
		return nil
	}
	for _, v := range list {
		ids = append(ids, v.Id)
		var monline = entity.CommDeviceOnline{
			Id:           id.NewId(),
			CommDeviceId: v.Id,
			OnlineDt:     datetime.FormatDateTime(time.Time(*v.LstRevDt)),
			OffDt:        "",
		}
		dataonline = append(dataonline, monline)
	}

	tx = db.DbMaster.Model(&entity.CommDevice{}).
		Where("id in ?", ids).
		Update("logic_state", "在线")
	if tx.Error != nil {
		logger.Error(tx.Error)
		return tx.Error
	} else {
		logger.Infof("在线设备数:%v", tx.RowsAffected)
	}

	tx = db.DbMaster.Create(dataonline)
	if tx.Error != nil {
		logger.Error(tx.Error)
		return tx.Error
	}
	return nil
}

func (p *OnlineServer) Offline(dbnow time.Time) error {
	if p.RunTimes%2 != 1 {
		return nil //10秒1次离线
	}

	var list []CommDeviceOnlineDto
	tx := db.DbMaster.Model(&entity.CommDeviceOnline{}).
		Joins("inner join CommDevice b on CommDeviceOnline.off_dt ='' and  CommDeviceOnline.comm_device_id =b.id ").
		Joins("inner join CommPort cp on b.comm_port_id =cp.id").
		Select(`CommDeviceOnline.id,
		CommDeviceOnline.online_dt ,
		CommDeviceOnline.comm_device_id ,
		b.lst_rev_dt ,
		b.logic_time_out ,
		cp.logic_time_out  as cp_logic_time_out`).
		Find(&list)
	if tx.Error != nil {
		logger.Error(tx.Error)
		return tx.Error
	}
	if len(list) == 0 {
		return nil
	}
	var ids []int64
	var updateList []entity.CommDeviceOnline
	for index, _ := range list {
		v := &list[index]
		var logic_time_out = v.LogicTimeOut
		if logic_time_out == 0 {
			logic_time_out = v.CpLogicTimeOut
		}
		var senconds = int32(time.Since(v.LstRevDt).Seconds())
		if senconds > logic_time_out {
			ids = append(ids, v.CommDeviceId)
			var OnlineDt, _ = datetime.ParseDateTime(v.OnlineDt)
			var mupdate = entity.CommDeviceOnline{
				Id:      v.Id,
				OffDt:   datetime.FormatDateTime(v.LstRevDt),
				Seconds: int(v.LstRevDt.Sub(OnlineDt).Seconds()),
			}
			updateList = append(updateList, mupdate)
		}
	}
	if len(ids) == 0 {
		return nil
	}
	var upmap = map[string]interface{}{
		"physical_state": "离线",
		"logic_state":    "离线",
	}
	tx = db.DbMaster.Model(&entity.CommDevice{}).Where("id in ?", ids).
		Updates(upmap)
	if tx.Error != nil {
		return tx.Error
	} else {
		logger.Infof("离线设备数:%v", tx.RowsAffected)
	}
	var fields = []string{"off_dt", "seconds"}
	tx = db.DbMaster.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "id"}},
		DoUpdates: clause.AssignmentColumns(fields),
	}).Create(&updateList)

	return nil
}
func (p *OnlineServer) Save(pkgs []*entity.StdPkg) {
	var revdata = make([]*entity.CommDevice, 0)
	var offline = make([]*entity.CommDevice, 0)
	for _, v := range pkgs {
		if v.PhysicalState == "离线" {
			var mold = p.FindDevice(offline, v.Sid)
			if mold != nil {
				mold.Id = v.Id //记录大id
				continue
			}
			var m = &entity.CommDevice{
				Id:  v.Id,
				Sid: v.Sid,
			}
			offline = append(offline, m)
		} else {
			var mold = p.FindDevice(revdata, v.Sid)
			if mold != nil {
				mold.Id = v.Id
				continue
			}
			var m = &entity.CommDevice{
				Id:            v.Id,
				Sid:           v.Sid,
				CommPortId:    v.CommPortId,
				CommProxyId:   v.CommProxyId,
				CommDealId:    v.CommDealId,
				ClientId:      v.ClientId,
				PhysicalState: "在线",
				LogicState:    "在线",
				LstRevDt:      nil,
				LogicTimeOut:  0,
			}
			var dt = datetime.Now()
			m.LstRevDt = &dt
			revdata = append(revdata, m)
		}
	}
	//删除离线信息
	for _, v := range revdata {
		var old = p.FindDevice(offline, v.Sid)
		if old != nil {
			//比较时间
			if old.Id > v.Id {
				v.PhysicalState = "离线"
				old.Sid = "" //不处理
			}
		}
	}
	var fields = []string{"comm_port_id", "comm_proxy_id", "comm_deal_id", "client_id", "physical_state", "logic_state", "lst_rev_dt"}
	tx := db.DbMaster.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "sid"}},
		DoUpdates: clause.AssignmentColumns(fields),
	}).Create(&revdata)
	logger.Infof("数据库操作 %v", tx)

	if len(offline) == 0 {
		return
	}
	var offsid = make([]string, 0)
	for _, v := range offline {
		if v.Sid != "" {
			offsid = append(offsid, v.Sid)
		}
	}
	if len(offsid) > 0 {
		tx = db.DbMaster.Model(&entity.CommDevice{}).Where("sid in ?", offsid).UpdateColumn("physical_state", "离线")
		logger.Info(tx)
	}

}
func (p *OnlineServer) FindDevice(arr []*entity.CommDevice, sid string) *entity.CommDevice {
	for _, v := range arr {
		if v.Sid == sid {
			return v
		}
	}
	return nil
}
